package com.atguigu.edu.realtime.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @ClassName: FixVisitorProcessFunction
 * @Description: TODO 访客状态修复处理函数
 * @Author: zhaoxunfeng
 * @Date: 2022-08-31 13:36
 * @Version: 1.0.0
 */
public class FixVisitorProcessFunction extends KeyedProcessFunction<String, JSONObject, JSONObject> {

    private ValueState<String> lastDayState;

    @Override
    public void open(Configuration parameters) throws Exception {
        lastDayState = getRuntimeContext()
                .getState(new ValueStateDescriptor<String>("last_day", String.class));
    }

    @Override
    public void processElement(JSONObject json,
                               Context ctx,
                               Collector<JSONObject> out) throws Exception {
        JSONObject common = json.getJSONObject("common");
        String isNew = common.getString("is_new");
        String lastDay = lastDayState.value();
        String currentDate = DateFormatUtil.toDateString(json.getLong("ts"));

        if ("1".equals(isNew)) { // 如果is_new的值为1 (新访客)
            if (lastDay == null) {
                lastDayState.update(currentDate);
            } else if (!lastDay.equals(currentDate)) {
                common.put("is_new", "0");
            }
        } else {
            if (lastDay == null) {
                lastDayState.update(DateFormatUtil.minusDay(currentDate, 1));
            }
        }
        out.collect(json);
    }
}
